package com.atguigu.gmall.realtime.app.dim;/**
 * Author lzc
 * Date 2022/4/29 8:56 上午
 */

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseAppV1;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.FlinkSinkUtil;
import com.atguigu.gmall.realtime.util.JdbcUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;


/**
 * @Author lzc
 * @Date 2022/4/29 8:56 上午
 */
public class DimApp extends BaseAppV1 {
    public static void main(String[] args) {
        new DimApp().init(
            "DimApp",
            2001,
            1,
            "DimApp",
            "DimApp",
            "ods_db"
        );
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       DataStreamSource<String> stream) {
        // 1. 对数据进行 etl
        SingleOutputStreamOperator<JSONObject> dataStream = eltDataStream(stream);
        // 2. 读取配置表数据
        SingleOutputStreamOperator<TableProcess> tableProcessStream = readTableProcess(env);
        // 3. 数据流和配置流进行 connect
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connectedStream = connect(dataStream, tableProcessStream);
        
        // 4. 过滤掉需要的字段
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> resultStream = filterNoSinkColumns(connectedStream);
        // 4. 写出到 Phoenix 中
        resultStream.addSink(FlinkSinkUtil.getPhoenixSink());
    }
    
    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> filterNoSinkColumns(
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> stream) {
        return stream.map(new MapFunction<Tuple2<JSONObject, TableProcess>, Tuple2<JSONObject, TableProcess>>() {
            @Override
            public Tuple2<JSONObject, TableProcess> map(Tuple2<JSONObject, TableProcess> t) throws Exception {
                List<String> needSinkColumns = Arrays.asList(t.f1.getSink_columns().split(","));
                
                t.f0.keySet().removeIf(c -> !needSinkColumns.contains(c));
                
                return t;
            }
        });
        
    }
    
    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connect(
        SingleOutputStreamOperator<JSONObject> dataStream,
        SingleOutputStreamOperator<TableProcess> tableProcessStream) {
        // 1. 把配置流做成广播流
        MapStateDescriptor<String, TableProcess> tpStateDesc = new MapStateDescriptor<>("tpState", String.class, TableProcess.class);
        
        BroadcastStream<TableProcess> tpBCStream = tableProcessStream.broadcast(tpStateDesc);
        
        
        // 2. 数据流 connect 广播流
        return dataStream
            .connect(tpBCStream)
            .process(new BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<JSONObject, TableProcess>>() {
                
                private Connection phoenixConn;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    phoenixConn = JdbcUtil.getJdbcConnection(Constant.PHOENIX_DRIVER, Constant.PHOENIX_URL, null, null);
                    
                }
                
                @Override
                public void processElement(JSONObject jsonObject,
                                           ReadOnlyContext ctx,
                                           Collector<Tuple2<JSONObject, TableProcess>> collector) throws Exception {
                    // 1. 获取广播该条数据对应的配置信息
                    ReadOnlyBroadcastState<String, TableProcess> tpState = ctx.getBroadcastState(tpStateDesc);
                    // 2. 把该条数据与对应的配置项做成Tuple2, 放入流中, 进行后序处理
                    String sourceTable = jsonObject.getString("table");
                    TableProcess tp = tpState.get(sourceTable);
                    if (tp != null) {
                        collector.collect(Tuple2.of(jsonObject.getJSONObject("data"), tp));
                    }
                }
                
                @Override
                public void processBroadcastElement(TableProcess tp,
                                                    Context ctx,
                                                    Collector<Tuple2<JSONObject, TableProcess>> collector) throws Exception {
                    // 1. 根据配置信息, 在 Phoenix 中建表
                    checkTable(tp);
                    // 2. 把配置信息放入到广播状态
                    BroadcastState<String, TableProcess> tpState = ctx.getBroadcastState(tpStateDesc);
                    tpState.put(tp.getSource_table(), tp);
                    
                }
                
                private void checkTable(TableProcess tp) throws SQLException {
                    
                    // create table if not exists user_info("id" varchar, name varchar, constraint pk primary key(id)) SALT_BUCKETS = 4;
                    StringBuilder sql = new StringBuilder();
                    sql
                        .append("create table if not exists ")
                        .append(Constant.PHOENIX_SCHEMA)
                        .append(".")
                        .append(tp.getSink_table())
                        .append("(")
                        .append(tp.getSink_columns().replaceAll("([^,]+)", "$1 varchar"))
                        .append(",constraint pk primary key(")
                        .append(tp.getSink_pk() == null ? "id" : tp.getSink_pk())
                        .append("))")
                        .append(tp.getSink_extend() == null ? "" : tp.getSink_extend());
                    
                    System.out.println("建表语句: " + sql.toString());
                    PreparedStatement ps = phoenixConn.prepareStatement(sql.toString());
                    ps.execute();
                    phoenixConn.commit();
                    ps.close();
                }
            });
        
        
    }
    
    private SingleOutputStreamOperator<TableProcess> readTableProcess(StreamExecutionEnvironment env) {
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
            .hostname("hadoop162")
            .port(3306)
            .databaseList("gmall_config")
            .tableList("gmall_config.table_process") // set captured tables [product, user, address ,order, custom]
            .username("root")
            .password("aaaaaa")
            .startupOptions(StartupOptions.initial())
            .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
            .build();
        
        return env
            .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysqlSource")
            .map(json -> {
                JSONObject obj = JSON.parseObject(json);
                return obj.getObject("after", TableProcess.class);
            });
        
    }
    
    private SingleOutputStreamOperator<JSONObject> eltDataStream(DataStreamSource<String> stream) {
        return stream
            .filter(json -> {
                try {
                    JSONObject obj = JSON.parseObject(json);
                    String type = obj.getString("type");
                    return obj.getJSONObject("data") != null
                        && ("insert".equals(type) || "update".equals(type) || "bootstrap-insert".equals(type));
                } catch (Exception e) {
                    System.out.println("错误格式数据....");
                    return false;
                }
            })
            .map(JSON::parseObject);
    }
}
